package com.rpc.core.remoting.impl;

import com.rpc.core.RpcService;
import com.rpc.core.codec.RpcCodec;
import com.rpc.core.codec.adapter.NettyDecodeAdapter;
import com.rpc.core.codec.adapter.NettyEncodeAdapter;
import com.rpc.core.constant.BeatConstant;
import com.rpc.core.param.RpcParams;
import com.rpc.core.holder.transmit.Request;
import com.rpc.core.holder.transmit.Response;
import com.rpc.core.remoting.Server;
import io.netty.bootstrap.ServerBootstrap;
import io.netty.channel.*;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.SocketChannel;
import io.netty.channel.socket.nio.NioServerSocketChannel;
import io.netty.handler.timeout.IdleStateEvent;
import io.netty.handler.timeout.IdleStateHandler;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.context.ApplicationContext;

import java.util.UUID;
import java.util.concurrent.*;

@ChannelHandler.Sharable
public class NettyService extends SimpleChannelInboundHandler<Request> implements Server {
    private Logger logger = LoggerFactory.getLogger(NettyService.class);
    private ApplicationContext context;

    private RpcService service;

    RpcParams params;

    public void setService(RpcService service) {
        this.service = service;
        this.context = service.getContext();
        this.params = this.context.getBean(RpcParams.class);
    }

    private ThreadPoolExecutor serverPool;

    Thread thread;


    @Override
    public void channelActive(ChannelHandlerContext ctx) throws Exception {
        logger.info("{} new channel",ctx.channel().id());
    }

    @Override
    public void start() {

        Object bean = context.getBean(params.getCodecType());

        if (!(bean instanceof RpcCodec)) {
            throw new RuntimeException("The specified encoder was not found!");
        }

        RpcCodec codec = (RpcCodec) bean;

        initThreadPool();

        thread = new Thread(() -> {

            // param
            EventLoopGroup bossGroup = new NioEventLoopGroup();
            EventLoopGroup workerGroup = new NioEventLoopGroup();

            try {
                // start server
                ServerBootstrap bootstrap = new ServerBootstrap();
                bootstrap.group(bossGroup, workerGroup)
                        .channel(NioServerSocketChannel.class)
                        .childHandler(new ChannelInitializer<SocketChannel>() {
                            @Override
                            public void initChannel(SocketChannel channel) throws Exception {
                                channel.pipeline()
                                        .addLast(new IdleStateHandler(0, 0, BeatConstant.BEAT_INTERVAL * 3, TimeUnit.SECONDS))     // beat 3N, close if idle
                                        .addLast(new NettyDecodeAdapter(codec))
                                        .addLast(new NettyEncodeAdapter(codec))
                                        .addLast(NettyService.this);
                            }
                        })
                        .childOption(ChannelOption.TCP_NODELAY, true)
                        .childOption(ChannelOption.SO_KEEPALIVE, true);

                // bind
                ChannelFuture future = bootstrap.bind(params.getPort()).sync();

                logger.info("rpc remoting server start success on port = {}", params.getPort());

                // wait util stop
                future.channel().closeFuture().sync();

            } catch (Exception e) {
                if (e instanceof InterruptedException) {
                    logger.info("rpc remoting server stop.");
                } else {
                    logger.error("rpc remoting server error.", e);
                }
//                throw new RuntimeException(e);
            } finally {

                try {
                    serverPool.shutdown();    // shutdownNow
                } catch (Exception e) {
                    logger.error(e.getMessage(), e);
                }
                try {
                    workerGroup.shutdownGracefully();
                    bossGroup.shutdownGracefully();
                } catch (Exception e) {
                    logger.error(e.getMessage(), e);
                }

            }
        });
        thread.setDaemon(true);
        thread.start();

    }

    @Override
    public void stop() {
        if (thread != null && thread.isAlive()) {
            thread.interrupt();
        }

        logger.info("rpc remoting server destroy success.");
    }

    @Override
    protected void channelRead0(ChannelHandlerContext ctx, Request request) throws Exception {

        Response responseBeat = checkBeat(request);

        if(responseBeat!=null){
            ctx.writeAndFlush(responseBeat);
            return;
        }

        // do invoke
        try {
            serverPool.execute(() -> {
                // 调用服务
                Response response = service.invoke(request);
                ctx.writeAndFlush(response);
            });
        } catch (Exception e) {
            // catch error
            Response responseError = new Response();
            responseError.setId(request.getId());
            responseError.setInvokeSuccess(false);
            responseError.setThrowable(e);

            ctx.writeAndFlush(responseError);
        }

    }

    private Response checkBeat(Request request) {
        if (BeatConstant.BEAT_ID == (request.getId())) {
            logger.info("rpc provider netty server read beat-ping.");
            return BeatConstant.BEAT_PONG;
        }
        return null;
    }

    private void initThreadPool() {
        this.serverPool = new ThreadPoolExecutor(
                params.getCorePoolSize(),
                params.getMaxPoolSize(),
                60L,
                TimeUnit.SECONDS,
                new LinkedBlockingQueue<Runnable>(1000),
                new ThreadFactory() {
                    @Override
                    public Thread newThread(Runnable r) {
                        return new Thread(r, "server pool" + UUID.randomUUID().toString());
                    }
                },
                new RejectedExecutionHandler() {
                    @Override
                    public void rejectedExecution(Runnable r, ThreadPoolExecutor executor) {
                        throw new RuntimeException(" Thread pool is EXHAUSTED!");
                    }
                });
    }


    @Override
    public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) {
        logger.error("rpc provider netty server caught exception", cause);
        ctx.close();
    }

    @Override
    public void userEventTriggered(ChannelHandlerContext ctx, Object evt) throws Exception {
        if (evt instanceof IdleStateEvent) {
            ctx.channel().close();
            logger.info("rpc provider netty server close an idle channel.");
        } else {
            super.userEventTriggered(ctx, evt);
        }
    }
}
